package okex

import (
	"bytes"
	"compress/flate"
	"encoding/json"
	"github.com/morgine/log"
	"github.com/shopspring/decimal"
	"io"
	"io/ioutil"
	"market/pkg/ws"
	"strconv"
	"strings"
	"time"
)

const (
	TradeTable   Table = "spot/trade"
	Depth5Table  Table = "spot/depth5"
	Depth12Table Table = "spot/depth_l2_tbt"
)

func initSubjectsMessage(symbols []string, tables ...Table) (subscribe, unsubscribe []byte) {
	sub := &subject{
		Op:   "subscribe",
		Args: nil,
	}
	for _, table := range tables {
		for _, symbol := range symbols {
			sub.Args = append(sub.Args, string(table)+":"+strings.ToUpper(symbol))
		}
	}
	subscribe, _ = json.Marshal(sub)
	sub.Op = "unsubscribe"
	unsubscribe, _ = json.Marshal(sub)
	return
}

func SubscribeDepth5(instruments []string, handle func(data []*Depth5Data)) (cancel func() error) {
	subscribe, unsubscribe := initSubjectsMessage(instruments, Depth5Table)
	return subscribeWs(subscribe, unsubscribe, func(msg *Message) {
		data, err := msg.Depth5Data()
		if err != nil {
			log.Error.Println(err)
		} else {
			handle(data)
		}
	})
}

type subject struct {
	Op   string   `json:"op"`
	Args []string `json:"args"`
}

type Table string

type Action string

type Message struct {
	Table  Table           `json:"table"`
	Action Action          `json:"action"`
	Data   json.RawMessage `json:"data"`
}

func (m *Message) Depth5Data() ([]*Depth5Data, error) {
	var dms []*Depth5Data
	err := json.Unmarshal(m.Data, &dms)
	if err != nil {
		return nil, err
	}
	return dms, nil
}

type Depth5Data struct {
	InstrumentID string         `json:"instrument_id"`
	Asks         []Depth5Orders `json:"asks"`
	Bids         []Depth5Orders `json:"bids"`
	Timestamp    time.Time      `json:"timestamp"`
}

type Depth5Orders []string

func (os Depth5Orders) Price() decimal.Decimal {
	p, _ := decimal.NewFromString(os[0])
	return p
}

func (os Depth5Orders) Size() decimal.Decimal {
	amount, _ := decimal.NewFromString(os[1])
	orders, _ := decimal.NewFromString(os[2])
	return amount.Mul(orders)
}

func (os Depth5Orders) FloatSize() float64 {
	amount, _ := strconv.ParseFloat(os[1], 64)
	orders, _ := strconv.Atoi(os[2])
	return amount * float64(orders)
}

func subscribeWs(connectMsgs, closeMsgs []byte, handler func(msg *Message)) (cancelFunc func() error) {
	wsClient := ws.NewClient("wss://real.okex.com:8443/ws/v3", func(w io.Writer) {
		_, err := w.Write(connectMsgs)
		if err != nil {
			log.Error.Println(err)
		}
	}, func(message []byte) {
		var err error
		message, err = ioutil.ReadAll(flate.NewReader(bytes.NewReader(message)))
		if err != nil {
			log.Error.Println(err)
		} else {
			if len(message) != 4 || string(message) != "pong" {
				msg := &Message{}
				err = json.Unmarshal(message, msg)
				if err != nil {
					log.Error.Println(err)
				} else {
					if len(msg.Data) == 0 {
						log.Info.Printf("okex: %s\n", message)
					} else {
						handler(msg)
					}
				}
			}
		}
	})

	cancel := make(chan struct{})

	go func() {
		ticker := time.NewTicker(25 * time.Second)
		for {
			select {
			case <-cancel:
				return
			case <-ticker.C:
				_, err := wsClient.Write([]byte("ping"))
				if err != nil {
					log.Error.Println(err)
				}
			}
		}
	}()
	wsClient.Run()
	return func() error {
		wsClient.Write(closeMsgs)
		cancel <- struct{}{}
		return wsClient.Close()
	}
}
